package plgnet

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/goutils/yhttp"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
	"io/ioutil"
	"strings"
	"sync/atomic"
	"time"
)

type PlgCacheHttpClt struct {
	startflag uint32
	pushfail  uint32

	plgbase.PlgBase
	url        string
	postCache  *utils.QueueMax // 非线程安全
	cacheQueue *utils.QueueTask

	lastPostData string
	lastPostResp string
	lastPostT    time.Time
	lastPostUrl  string
	lastPostInfo string
}

/*
**

	返回true, 表示处理完成(可接受错误).
	false: 意外错误
*/
func (this *PlgCacheHttpClt) innerPost(url string, json *wrapper.SuperValue) bool {

	strBody := json.AsJSONString(false)
	this.lastPostData = strBody
	this.lastPostT = time.Now()

	if this.OutputEx != nil {
		this.OutputEx.CheckGetPlug()
		if this.OutputEx.IsValid() {
			this.lastPostUrl = this.OutputEx.StatusString()
			this.OutputEx.SendJSON("", json)
			return true
		}
	}

	this.lastPostUrl = url
	if len(url) == 0 {
		return true
	}

	t1 := time.Now()
	defer func() {
		this.lastPostInfo = fmt.Sprintf("[%s]consume:%.2f (s)", utils.DateTimeStringForShortShow(time.Now()), time.Since(t1).Seconds())
	}()
	resp, err := yhttp.DefaultHttpPost(url, "application/json", strings.NewReader(strBody))
	if err != nil {
		this.lastPostResp = fmt.Sprintf("[%s]提交数据出现了异常:%s", utils.NowString(), err.Error())
		return false
	}
	defer resp.Body.Close()
	if resp.StatusCode == 404 || resp.StatusCode == 503 {
		this.lastPostResp = fmt.Sprintf("[%s]响应异常[%d]:%s", utils.NowString(), resp.StatusCode, PrjHttpUtils.GetRequestHeaderAsString(resp.Header))
		return false
	}

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		this.lastPostResp = PrjHttpUtils.GetRequestHeaderAsString(resp.Header)
		golog.Warnf("提交数据后,读取响应异常:%s", err.Error())
		return true
	}

	this.lastPostResp = string(body)
	if len(this.lastPostResp) > 0 {
		retJson, err := wrapper.NewSVFromBuf(body)
		if err != nil {
			return true
		}

		if retJson == nil {
			return true
		}

		if retJson.IntByName("code", 0) == -1 {
			golog.Warnf("提交数据后,响应异常: %s", json.AsJSONString(false))
		}
	}
	return true
}

func (this *PlgCacheHttpClt) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	var rb utils.BytesBuilder
	if this.OutputEx != nil {
		rb.Appendf("output:%s", this.OutputEx.StatusString())
	}
	rb.Appendf("cache(raw in):%d, pushfail:%d\r\n", this.cacheQueue.Count(), this.pushfail)
	rb.Appendf("cache(pacakge):%d, max:%d\r\n", this.postCache.Count(), this.postCache.GetMaxQueueSize())
	rb.Appendf("%s\r\n", this.lastPostInfo)
	rb.Appendf("time:%s, url:%s\r\n================================\r\n", utils.DateTimeString2(this.lastPostT), this.lastPostUrl)
	rb.AppendStr("post\r\n")
	rb.AppendStr(this.lastPostData).AppendStr("\r\n")
	rb.AppendStr("\r\n\r\n")
	rb.AppendStr("response(响应如果出现异常, 则会中断, 而且会进行重试)\r\n")
	rb.AppendStr(this.lastPostResp)

	rec.RespW.Write(rb.Bytes())
	return
}

/*
**
只能单线程执行, 使用了Peek
*/
func (this *PlgCacheHttpClt) checkPost() {
	url := plgbase.ReplacePlaceholderForFileName(this.url)

	for {
		ok, val := this.postCache.Peek()
		if ok {
			json := val.(*wrapper.SuperValue)
			if json == nil {
				break
			}
			json.SetKeyValue("t", time.Now().Unix())
			if this.innerPost(url, json) {
				this.postCache.Pop() // 完成
				json.Clear()         // 清理
			} else {
				break // 如果出现意外错误将终止执行
			}
		} else {
			break
		}
	}
}

func (this *PlgCacheHttpClt) OnTaskWork() {

	json := wrapper.NewSVObjectCache(true)
	lst := json.ForceArrayByPath("list")
	n := 0
	this.cacheQueue.PopTaskArgMaxFunc(10, func(args ...interface{}) bool {
		if len(args) == 0 {
			golog.Warnf("压入了空数据!!!")
			return true
		}
		if r1, ok := args[0].(*wrapper.SuperValue); ok {
			lst.ArrayAppendValue(r1)
			n++
		} else {
			golog.Warnf("非法的数据类型!!!")
		}
		return true
	})

	if n > 0 { // 压入缓存中
		this.postCache.Push(json)
	} else {
		json.Clear()
	}
	this.checkPost()

}

/*
**

	{
	  "id":"cache-post",
	  "type":"core.http.cache.clt",
	  "conf":{
	    "input-ex":{"id":"ex", "topic":"jt"},
		"url":"http://xxx.xx.xx:xxx/$appuuid$,
	    "output":{},
		"maxcache":1024,
		"idle_sleep_interval_secs":10   // 空闲时sleep时间 默认 1 ms
	   }
	}

maxcache: 缓存队列最大容量, 压入时超过会产生pushfail

* 会打包一起提交, 提交失败会重新提交. 单线程提交数据
*/
func (this *PlgCacheHttpClt) ReloadConf(conf *wrapper.SuperValue) error {
	this.ReloadInputConf(conf.ValueByName("input-ex"), func(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
		this.SendKVs(channelid, data)
		return true
	})

	if this.postCache == nil {
		this.postCache = utils.NewQueueMax()
		this.postCache.ConfigMaxQueueSize(5)
	}

	if atomic.CompareAndSwapUint32(&this.startflag, 0, 1) {
		this.cacheQueue = utils.NewQueueTaskFunc()
		this.cacheQueue.ConfigMaxNum(1)
		this.cacheQueue.ConfigIntervalCallBack(time.Second, this.OnTaskWork)
		this.cacheQueue.Start()
	}

	conf.CheckExistsStrFunc("url", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		this.url = plgbase.ReplacePlaceholderForFileName(strv)
	})

	conf.CheckExistsIntFunc("maxcache", func(val int64) {
		this.cacheQueue.ConfigMaxQueueSize(int(val))
	})

	conf.CheckExistsIntFunc("idle_sleep_interval_secs", func(val int64) {
		this.cacheQueue.ConfigIdleSleepTime(time.Duration(val) * time.Second)
	})

	this.ReloadOutputConf(conf.ValueByName("output"))

	return nil
}

func (this *PlgCacheHttpClt) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.ReleaseFlag = 1
		this.cacheQueue.Stop()
	}
	return
}

func (this *PlgCacheHttpClt) SendKVs(topic string, kv ...interface{}) error {
	if this.ReleaseFlag == 1 {
		return nil
	}
	if this.cacheQueue != nil {
		l := len(kv)
		if l == 1 {
			if json, ok := kv[0].(*wrapper.SuperValue); ok {
				err := this.cacheQueue.Push(json.CloneCache(true))
				if err != nil {
					atomic.AddUint32(&this.pushfail, 1)
				}
				return nil // 输出日志太多错误
			}
		}
		json := wrapper.NewSVObjectCache(true)
		json.SetKeyValue("topic", topic)
		json.SetKeyValues(kv...)
		err := this.cacheQueue.Push(json)
		if err != nil {
			atomic.AddUint32(&this.pushfail, 1)
		}
		return nil // 输出日志太多错误
	} else {
		return fmt.Errorf("插件尚未进行初始化")
	}
}

func CreatePlgCacheHttpCltFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgCacheHttpClt{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
